from pyflink.common import Duration, WatermarkStrategy
from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode
from pyflink.datastream.connectors.file_system import FileSource, StreamFormat

# 1、创建flink执行环境
env = StreamExecutionEnvironment.get_execution_environment()

# 修改处理模型  （流批一体）
# BATCH: 批处理模型
# 1、输出最终结果
# 2、只能用于处理有界流
# 3、底层时mapreduce模型
# STREAMING：流处理模型（默认）
# 1、输出中间结果
# 2、流处理可以处理有界流和无界流, 处理有界流输出处理完就结束，处理无界流任务就一直运行
# 3、底层时持续流模型
env.set_runtime_mode(RuntimeExecutionMode.STREAMING)

# 修改并行度
env.set_parallelism(200)


# 设置数据发送缓存的超时时间
env.set_buffer_timeout(0)

# 2、监控目录构建无界流
# for_record_stream_format: 指定读取文件的格式和路径
# monitor_continuously：定时监控目录
file_source = FileSource \
    .for_record_stream_format(StreamFormat.text_line_format(), "G:\LanZhiPeiXun\Flink\data\words.txt") \
    .monitor_continuously(Duration.of_seconds(1)) \
    .build()

# 基于file source构建DS(无界流)
lines_ds = env.from_source(file_source, WatermarkStrategy.no_watermarks(), "FileSource")

# 统计单词的数量
# 一行转换成多行
words_ds = lines_ds.flat_map(lambda line: line.split(","))
# 转换成kv格式
kv_ds = words_ds.map(lambda word: (word, 1))
# 按照单词分组
key_by_ds = kv_ds.key_by(lambda kv: kv[0])
# 统计单词的数量
count_ds = key_by_ds.reduce(lambda x, y: (y[0], x[1] + y[1]))

count_ds.print()

# 启动flink任务
env.execute()
